"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 可靠性-数据库系统类
Case Name   : 开启归档模式，pg_xlog日志数量超过checkpoint_segments设定的值时日志归档成功
Create At   : 2022/6/20
Owner       : @li-xin12345
Description :
    1.修改参数将日志归档模式打开
    2.配置归档路径，不断执行sql脚本在archive_status目录下刷出.done文件
    3.查看归档日志模式
    4.查看日志是否正常归档
    5.删除表，恢复参数
Expect      :
    1.修改参数成功
    2.配置归档路径成功，执行事务.done文件成功生成
    3.归档日志模式正常
    4.WAL日志正常归档
    5.删除表成功，恢复参数成功
History     :
    Modified by wx1115623 2022/8/02:修改创建目录命令
"""
import os
import time
import unittest
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import RestartDbCluster
from yat.test import Node
from yat.test import macro


class Dbsys(unittest.TestCase):
    nodes_tuple = ('PrimaryDbUser', 'Standby1DbUser', 'Standby2DbUser')

    @RestartDbCluster(*nodes_tuple)
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.primary_sh = CommonSH('PrimaryDbUser')
        self.common = Common()
        self.constant = Constant()
        self.dirpath = os.path.join(macro.ARCH_PATH, 'arch_dbsys_case051')
        self.table_name = 't_dbsys_case051'
        
        archive_mode_msg = self.primary_sh.executDbSql('show archive_mode;')
        self.logger.info(archive_mode_msg)
        self.archive_mode_default = archive_mode_msg.splitlines()[-2].strip()

        archive_timeout_msg = self.primary_sh.executDbSql(
            'show archive_timeout;')
        self.logger.info(archive_timeout_msg)
        self.archive_timeout_default = \
            archive_timeout_msg.splitlines()[-2].strip()

        checkpoint_segments_msg = self.primary_sh.executDbSql(
            'show checkpoint_segments;')
        self.logger.info(checkpoint_segments_msg)
        self.checkpoint_segments_default = \
            checkpoint_segments_msg.splitlines()[-2].strip()

        archive_command_msg = self.primary_sh.executDbSql(
            'show archive_command;')
        self.logger.info(archive_command_msg)
        self.archive_command_default = \
            archive_command_msg.splitlines()[-2].strip()

    def test_dbsys(self):
        text = '-----step1:修改参数将日志归档模式打开;expect:修改参数成功-----'
        self.logger.info(text)
        result = self.primary_sh.executeGsguc('reload',
                                              self.constant.GSGUC_SUCCESS_MSG,
                                              'archive_mode=on')
        self.assertTrue(result, '执行失败' + text)
        show_msg = self.primary_sh.executDbSql('show archive_mode;')
        self.logger.info(show_msg)
        self.assertEquals('on', show_msg.splitlines()[-2].strip(),
                          '执行失败' + text)
        result = self.primary_sh.executeGsguc('reload',
                                              self.constant.GSGUC_SUCCESS_MSG,
                                              'archive_timeout=60')
        self.assertTrue(result, '执行失败' + text)
        show_msg = self.primary_sh.executDbSql('show archive_timeout;')
        self.logger.info(show_msg)
        self.assertEquals('1min', show_msg.splitlines()[-2].strip(),
                          '执行失败' + text)
        result = self.primary_sh.executeGsguc('reload',
                                              self.constant.GSGUC_SUCCESS_MSG,
                                              'checkpoint_segments=16')
        self.assertTrue(result, '执行失败' + text)
        show_msg = self.primary_sh.executDbSql('show checkpoint_segments;')
        self.logger.info(show_msg)
        self.assertEquals('16', show_msg.splitlines()[-2].strip(),
                          '执行失败' + text)

        text = '-----step2:配置归档路径，不断执行sql脚本在archive_status目录下' \
               '刷出.done文件;expect:配置归档路径成功，执行事务.done文件成功生成-----'
        self.logger.info(text)
        self.logger.info('-----配置归档路径-----')
        mkdir_cmd = f"rm -rf {self.dirpath} && mkdir -p {self.dirpath}"
        self.logger.info(mkdir_cmd)
        mkdir_msg = self.primary_node.sh(mkdir_cmd).result()
        self.logger.info(mkdir_msg)
        check_dir_cmd = f'''if [ -d {self.dirpath} ]; 
            then echo "exists"; else echo "not exists"; fi'''
        self.logger.info(check_dir_cmd)
        mkdir_result = self.primary_node.sh(check_dir_cmd).result()
        self.logger.info(mkdir_result)
        self.assertEquals("exists", mkdir_result, '执行失败' + text)
        result = self.primary_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f"archive_command='cp --remove-destination %p {self.dirpath}/%f'")
        self.assertTrue(result, '执行失败' + text)
        show_msg = self.primary_sh.executDbSql('show archive_command;')
        self.logger.info(show_msg)
        self.assertIn(self.dirpath, show_msg.splitlines()[-2].strip(),
                      '执行失败' + text)
        stop_msg = self.primary_sh.stopDbCluster()
        self.logger.info(stop_msg)
        self.assertTrue(stop_msg, '停库失败')
        start_res = self.primary_sh.startDbCluster()
        self.logger.info(start_res)
        self.assertTrue(start_res, '启动失败')
        time.sleep(10)

        self.logger.info('-----查询原有的archive_status日志-----')
        archive_status_path = os.path.join(macro.PG_XLOG_PATH, 'archive_status')
        ls_archive_cmd = f'ls {archive_status_path}'
        self.logger.info(ls_archive_cmd)
        ls_archive_msg = self.primary_node.sh(ls_archive_cmd).result()
        self.logger.info(ls_archive_msg)
        old_done_num = len(ls_archive_msg.split())

        self.logger.info('-----不断执行sql脚本在archive_status目录下刷出.done文件-----')
        sql_cmd = f'''declare
            tb_name text;
            begin
                for i in 0..1000 loop
                    tb_name := '{self.table_name}_' || i;
                    execute 'drop table if exists ' || tb_name || ';';
                    execute 'create table ' || tb_name ||'(id int,name int);';
                    execute 'insert into '|| tb_name || ' 
                    values(generate_series(1,5000),generate_series(1,5000));';
                end loop;
            end;'''
        self.logger.info(sql_cmd)
        sql_msg = self.primary_sh.executDbSql(sql_cmd)
        self.logger.info(sql_msg)
        self.assertIn(self.constant.anonymous_block_success_msg,
                      sql_msg, '执行失败' + text)
        checkpoint_msg = self.primary_sh.executDbSql('checkpoint;')
        self.logger.info(checkpoint_msg)
        self.assertIn(self.constant.checkpoint_success_msg,
                      checkpoint_msg, '执行失败' + text)
        time.sleep(10)
        self.logger.info('-----查询.done文件是否成功生成-----')
        check_cmd = f"ls {archive_status_path}"
        self.logger.info(check_cmd)
        check_msg = self.primary_node.sh(check_cmd).result()
        self.logger.info(check_msg)
        new_done_num = len(check_msg.split())
        self.assertGreater(new_done_num, old_done_num, '执行失败' + text)

        text = '-----step3:查看归档日志模式;expect:归档日志模式正常-----'
        self.logger.info(text)
        show_msg = self.primary_sh.executDbSql('show archive_mode;')
        self.logger.info(show_msg)
        self.assertEquals('on', show_msg.splitlines()[-2].strip(),
                          '执行失败' + text)

        text = '-----step4:查看日志是否正常归档;expect:WAL日志正常归档-----'
        self.logger.info(text)
        self.logger.info('-----查询当前的pg_xlog日志-----')
        ls_cmd_new = f'ls {macro.PG_XLOG_PATH} | grep -v archive_status '
        self.logger.info(ls_cmd_new)
        ls_msg_new = self.primary_node.sh(ls_cmd_new).result()
        self.logger.info(ls_msg_new)
        new_xlog_num = len(ls_msg_new.splitlines())
        self.assertGreater(new_xlog_num, 16, '执行失败' + text)
        self.logger.info('-----查询归档日志-----')
        check_cmd = f"ls {self.dirpath}"
        self.logger.info(check_cmd)
        check_msg = self.primary_node.sh(check_cmd).result()
        self.logger.info(check_msg)
        self.assertGreater(len(check_msg.split()), 0, '执行失败' + text)

    def tearDown(self):
        text = '-----step5:删除表，恢复参数;expect:删除表成功，恢复参数成功-----'
        self.logger.info(text)
        drop_cmd = f'''declare
                    tb_name text;
                    begin
                        for i in 0..1000 loop
                            tb_name := '{self.table_name}_' || i;
                            execute 'drop table ' || tb_name || ';';
                        end loop;
                    end;'''
        self.logger.info(drop_cmd)
        drop_msg = self.primary_sh.executDbSql(drop_cmd)
        self.logger.info(drop_msg)

        result_1 = self.primary_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'archive_mode={self.archive_mode_default}')
        show_archive_mode = self.primary_sh.executDbSql('show archive_mode;')
        self.logger.info(show_archive_mode)

        result_2 = self.primary_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'archive_timeout={self.archive_timeout_default}')
        show_archive_timeout = self.primary_sh.executDbSql(
            'show archive_timeout;')
        self.logger.info(show_archive_timeout)

        result_3 = self.primary_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'checkpoint_segments={self.checkpoint_segments_default}')
        show_checkpoint = self.primary_sh.executDbSql(
            'show checkpoint_segments;')
        self.logger.info(show_checkpoint)

        result_4 = self.primary_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f"archive_command= 'cp --remove-destination %p "
            f"{self.archive_command_default}/%f'")
        show_archive_command = self.primary_sh.executDbSql(
            'show archive_command;')
        self.logger.info(show_archive_command)

        self.logger.info('-----断言tearDown-----')
        self.assertIn(self.constant.anonymous_block_success_msg,
                      drop_msg, '执行失败' + text)
        self.assertTrue(result_1, '执行失败' + text)
        self.assertEquals(self.archive_mode_default,
                          show_archive_mode.splitlines()[-2].strip(),
                          '执行失败' + text)
        self.assertTrue(result_2, '执行失败' + text)
        self.assertEquals(self.archive_timeout_default,
                          show_archive_timeout.splitlines()[-2].strip(),
                          '执行失败' + text)
        self.assertTrue(result_3, '执行失败' + text)
        self.assertEquals(self.checkpoint_segments_default,
                          show_checkpoint.splitlines()[-2].strip(),
                          '执行失败' + text)
        self.assertTrue(result_4, '执行失败' + text)
        self.assertIn(self.archive_command_default,
                      show_archive_command.splitlines()[-2].strip(),
                      '执行失败' + text)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
